package ar.edu.itba.pod.legajo47573;

import java.rmi.RemoteException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map.Entry;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

import org.jgroups.Address;
import org.jgroups.JChannel;
import org.jgroups.Message;
import org.jgroups.ReceiverAdapter;
import org.jgroups.View;

import ar.edu.itba.pod.api.NodeStats;
import ar.edu.itba.pod.api.Result;
import ar.edu.itba.pod.api.Result.Item;
import ar.edu.itba.pod.api.SPNode;
import ar.edu.itba.pod.api.Signal;
import ar.edu.itba.pod.api.SignalProcessor;
import ar.edu.itba.pod.legajo47573.messages.MainMessage;
import ar.edu.itba.pod.legajo47573.messages.MultipleSignalsMessage;
import ar.edu.itba.pod.legajo47573.messages.OneSignalMessage;
import ar.edu.itba.pod.legajo47573.messages.OwnerBackupSignal;
import ar.edu.itba.pod.legajo47573.messages.OwnerBackupSignalList;
import ar.edu.itba.pod.legajo47573.messages.ResultMessage;

public class NodeImpl extends ReceiverAdapter implements SPNode, SignalProcessor {
	/************************************************************************************************/
	/***********************************Instance variables and constructor***************************/
	private final int threadsNumber;
	
	/*Cluster*/
	private JChannel channel;
	private Address address;
	private BlockingQueue<Address> connectedNodes = new LinkedBlockingQueue<Address>();
	
	private BlockingQueue<SignalsToFind> signalsToFind = new LinkedBlockingQueue<SignalsToFind>();
	private BlockingQueue<Result> resultsQueue;
	private BlockingQueue<Result> resultList;
	private BlockingQueue<UpDownNode> viewAcceptedQueue = new LinkedBlockingQueue<UpDownNode>();
	private Address whoStartedSearch = null;

	/*Signals processing data structures*/
	/*Address is the address who backups my sublist of signals*/
	private ConcurrentHashMap<Address, List<OwnerBackupSignal>> mySignals = new ConcurrentHashMap<Address, List<OwnerBackupSignal>>();
	/*Address is the address the owner of the sublist of signals*/
	private ConcurrentHashMap<Address, List<OwnerBackupSignal>> signalsIBackup = new ConcurrentHashMap<Address, List<OwnerBackupSignal>>();
	private BlockingQueue<Signal> signalsNodeAlone = new LinkedBlockingQueue<Signal>();
	private BlockingQueue<Signal> arrivingSignalsQueue = new LinkedBlockingQueue<Signal>();
	private BlockingQueue<Message> receivedMessagesQueue = new LinkedBlockingQueue<Message>();
	
	/*Latches and semaphores*/
	private CountDownLatch latch;
	private CountDownLatch deprecatedLatch;
	private CountDownLatch distributeSignalsLatch;
	private CountDownLatch degradedLatch;
	private CountDownLatch newNodeLatch;
	private CountDownLatch latchToWaitForDegradation;
	private CountDownLatch degradedLatchForNewNode;
	private CountDownLatch distributingOnNewNodeLatch;
	
	/*Atomics*/
	private AtomicInteger comparisonSignals = new AtomicInteger();
	private AtomicBoolean imNew = new AtomicBoolean(true);
	private AtomicBoolean completed = new AtomicBoolean(false);
	private AtomicBoolean isInSearch = new AtomicBoolean(false);
	private AtomicBoolean degraded = new AtomicBoolean(false);
	private AtomicBoolean searchInterrupted = new AtomicBoolean(false);
	private AtomicBoolean distributing = new AtomicBoolean(false);
	
	/*Threads and executors*/
	private Thread viewAcceptedThread;
	private Thread dequeingSignalThread;
	private Thread receiveMessageThread;
	
	private Thread searchThread;
	private ExecutorService signalsExecutor;
	
	private ExecutorService threadsExecutor = Executors.newFixedThreadPool(3);
	
	public NodeImpl(int threadsNumber){
		this.threadsNumber = threadsNumber;
	}
	/************************************************************************************************/
	/************************************************************************************************/
	

	/************************************************************************************************/
	/*****************************************View accepted******************************************/
	
	@Override
	public void viewAccepted(View view) {
		List<Address> newConnectedNodes = view.getMembers();
		if(connectedNodes.isEmpty())
			imNew.set(true);
		for(Address node : newConnectedNodes){
			if(!connectedNodes.contains(node)){
				try {
					viewAcceptedQueue.put(new UpDownNode(node, ViewAcceptedMsg.UP));
					connectedNodes.put(node);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
		}
		for(Address node : connectedNodes){
			if(!newConnectedNodes.contains(node)){
				try {
					viewAcceptedQueue.put(new UpDownNode(node, ViewAcceptedMsg.DOWN));
					connectedNodes.remove(node);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
		}
		System.out.println("connected nodes: " + connectedNodes);
	}
	
	
	/************************************************************************************************/
	/*************************************Fallen node distribution***********************************/
	
	
	private synchronized void distributeSignals(final Address fallenNode){
		new Thread(){
			@Override
			public void run() {
				degraded.set(true);
				/*Useless, just to use the same message*/
				degradedLatchForNewNode = new CountDownLatch(1);
				/*Coordinator set everyone in degraded mode*/
				if(channel.getView().getMembers().get(0) == address)
					try {
						channel.send(null, new OneSignalMessage(MsgOptions.DEGRADED, null));
					} catch (Exception e1) {
						e1.printStackTrace();
					}
				System.out.println("Fallen node detected. Redistributing signals...");
				/*If a node falls down and there is only one left*/
				if(channel.getView().getMembers().size() == 1){
					List<OwnerBackupSignal> backupedSignalList = signalsIBackup.get(fallenNode);
					/*Get the backuped signals*/
					if(backupedSignalList == null)
						return;
					for(OwnerBackupSignal signal : backupedSignalList){
						/*Change the owner of the signal*/
						signal.setOwner(address);
						/*Add the signal to my signals*/
						if(mySignals.get(address) == null){
							List<OwnerBackupSignal> newList = new ArrayList<OwnerBackupSignal>();
							newList.add(signal);
							mySignals.put(address, newList);
						}
						else{
							List<OwnerBackupSignal> signalList = mySignals.get(address);
							signalList.add(signal);
						}
					}
					System.out.println("After");
					/*Transfer my signals that where in the fallen key to my key*/
					List<OwnerBackupSignal> signalListUpdate = mySignals.get(address);
					List<OwnerBackupSignal> signalListToErase = mySignals.get(fallenNode);
					signalListUpdate.addAll(signalListToErase);
					mySignals.remove(fallenNode);

					/*Backup the signals in mySignals*/
					List<OwnerBackupSignal> mySignalList = mySignals.get(address); 
					signalsIBackup.put(address, mySignalList);
					signalsIBackup.remove(fallenNode);
					degraded.set(false);
				}
				else{
					latchToWaitForDegradation = new CountDownLatch(1);
					int totalSignals = 0;
					ConcurrentHashMap<Address, OwnerBackupSignalList> signalsToSendOwn = new ConcurrentHashMap<Address, OwnerBackupSignalList>();
					ConcurrentHashMap<Address, OwnerBackupSignalList> signalsToSendBackup = new ConcurrentHashMap<Address, OwnerBackupSignalList>();
					if(signalsIBackup.containsKey(fallenNode)){
						List<OwnerBackupSignal> backupedSignalList = signalsIBackup.get(fallenNode);
						signalsIBackup.remove(fallenNode);

						for(OwnerBackupSignal signal : backupedSignalList){
							/*Redistribute signals of the fallen node*/
							List<Address> nodesList = new ArrayList<Address>();
							nodesList.addAll(channel.getView().getMembers());
							/*Choose a node to be the owner*/
							int random = (int)(Math.random() * (nodesList.size()));		

							Address ownerNodeAddress = nodesList.get(random);

							nodesList = new ArrayList<Address>();
							nodesList.addAll(channel.getView().getMembers());
							/*Choose a node to be the backup different from the owner*/
							if(nodesList.size() > 1)
								nodesList.remove(ownerNodeAddress);
							random = (int)(Math.random() * (nodesList.size()));

							Address backupNodeAddress = nodesList.get(random);

							OwnerBackupSignal ownerBackupSignal = new OwnerBackupSignal(signal.getSignal(), ownerNodeAddress, backupNodeAddress);

							//TODO put this on another method 
							/*The keys of the signalsToSend maps are the destination node of the message to send*/
							if(!signalsToSendOwn.containsKey(ownerNodeAddress)){
								OwnerBackupSignalList obsl = new OwnerBackupSignalList();
								obsl.addOwn(ownerBackupSignal);
								signalsToSendOwn.put(ownerNodeAddress, obsl);
							}
							else{
								OwnerBackupSignalList obsl = signalsToSendOwn.get(ownerNodeAddress);
								obsl.addOwn(ownerBackupSignal);
							}
							/*The same with the backups*/
							if(!signalsToSendOwn.containsKey(backupNodeAddress)){
								OwnerBackupSignalList obsl = new OwnerBackupSignalList();
								obsl.addBackup(ownerBackupSignal);
								signalsToSendOwn.put(backupNodeAddress, obsl);
							}
							else{
								OwnerBackupSignalList obsl = signalsToSendOwn.get(backupNodeAddress);
								obsl.addBackup(ownerBackupSignal);
							}
						}
					}
					if(mySignals.containsKey(fallenNode)){
						List<OwnerBackupSignal> ownedSignalList = mySignals.get(fallenNode);
						mySignals.remove(fallenNode);

						for(OwnerBackupSignal signal : ownedSignalList){
							/*Redistribute signals of the fallen node*/
							List<Address> nodesList = new ArrayList<Address>();
							nodesList.addAll(channel.getView().getMembers());
							/*Choose a node to be the owner*/
							int random = (int)(Math.random() * (nodesList.size()));		

							Address ownerNodeAddress = nodesList.get(random);

							nodesList = new ArrayList<Address>();
							nodesList.addAll(channel.getView().getMembers());
							/*Choose a node to be the backup different from the owner*/
							if(nodesList.size() > 1)
								nodesList.remove(ownerNodeAddress);
							random = (int)(Math.random() * (nodesList.size()));

							Address backupNodeAddress = nodesList.get(random);

							OwnerBackupSignal ownerBackupSignal = new OwnerBackupSignal(signal.getSignal(), ownerNodeAddress, backupNodeAddress);

							/*The keys of the signalsToSend maps are the destination node of the message to send*/
							if(!signalsToSendBackup.containsKey(ownerNodeAddress)){
								OwnerBackupSignalList obsl = new OwnerBackupSignalList();
								obsl.addOwn(ownerBackupSignal);
								signalsToSendBackup.put(ownerNodeAddress, obsl);
							}
							else{
								OwnerBackupSignalList obsl = signalsToSendBackup.get(ownerNodeAddress);
								obsl.addOwn(ownerBackupSignal);
							}
							/*The same with the backups*/
							if(!signalsToSendBackup.containsKey(backupNodeAddress)){
								OwnerBackupSignalList obsl = new OwnerBackupSignalList();
								obsl.addBackup(ownerBackupSignal);
								signalsToSendBackup.put(backupNodeAddress, obsl);
							}
							else{
								OwnerBackupSignalList obsl = signalsToSendBackup.get(backupNodeAddress);
								obsl.addBackup(ownerBackupSignal);
							}
						}
					}
					distributeSignalsLatch = new CountDownLatch(signalsToSendOwn.size()+signalsToSendBackup.size());
					if(channel.getView().getMembers().get(0) == address)
						degradedLatch = new CountDownLatch(channel.getView().getMembers().size());

					/*Now the lists are built and its time to send the messages*/
					Iterator<Entry<Address, OwnerBackupSignalList>> iteratorOwn = signalsToSendOwn.entrySet().iterator();
					while(iteratorOwn.hasNext()){
						Entry<Address, OwnerBackupSignalList> entry = iteratorOwn.next();
						try {
							channel.send(entry.getKey(), new MultipleSignalsMessage(MsgOptions.ADD_MULTIPLE_SIGNALS, entry.getValue()));
							totalSignals += entry.getValue().getOwnedSignals().size();
						} catch (Exception e) {
							e.printStackTrace();
						}
					}
					Iterator<Entry<Address, OwnerBackupSignalList>> iteratorBack = signalsToSendBackup.entrySet().iterator();
					while(iteratorBack.hasNext()){
						Entry<Address, OwnerBackupSignalList> entry = iteratorBack.next();
						try {
							channel.send(entry.getKey(), new MultipleSignalsMessage(MsgOptions.ADD_MULTIPLE_SIGNALS, entry.getValue()));
							totalSignals += entry.getValue().getBackupedSignals().size();
						} catch (Exception e) {
							e.printStackTrace();
						}
					}
					try {
						System.out.println("HERE AWAIT");
						distributeSignalsLatch.await();

						System.out.println("COORDINATOR " + channel.getView().getMembers().get(0));
						System.out.println("ME " + address);
						/*Tell the coordinator that I finished balancing*/
						try {
							channel.send(channel.getView().getMembers().get(0), new OneSignalMessage(MsgOptions.NOT_DEGRADED, null));
						} catch (Exception e) {
							e.printStackTrace();
						}
					} catch (InterruptedException e) {
						e.printStackTrace();
					}

					if(channel.getView().getMembers().get(0) == address){
						try {
							/*Wait for the nodes to respond*/
							degradedLatch.await(totalSignals * 3 + 100, TimeUnit.SECONDS);
							try {
								channel.send(null, new OneSignalMessage(MsgOptions.COORDINATOR_NOT_DEGRADED, null));
							} catch (Exception e) {
								e.printStackTrace();
							}
						} catch (InterruptedException e) {
							e.printStackTrace();
						}
					}

					try {
						latchToWaitForDegradation.await();
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
					degraded.set(false);
				}
			}
		}.start();
	}
	
	/************************************************************************************************/
	/************************************************************************************************/
	
	
	/************************************************************************************************/
	/****************************************New node distribution***********************************/
	/*destNode is the new node where signals will be sent*/
	private synchronized void sendSignalsToNewNode(final Address destNode) {
		new Thread(){
			@Override
			public void run() {

				System.out.println("Signal balancing started...");

				List<OwnerBackupSignal> mySignalsList = mapToList(mySignals); 
				int signalsSize = mySignalsList.size();
				int nodesInCluster = channel.getView().getMembers().size();
				int signalsToSendNumber = signalsSize-(signalsSize*(nodesInCluster-1))/nodesInCluster;

				System.out.println("SignalsToSendNumber " + signalsToSendNumber);

				System.out.println("Before");

				if( nodesInCluster > 1){
					/*Useless for new node*/
					distributeSignalsLatch = new CountDownLatch(1);
					degradedLatchForNewNode = new CountDownLatch(1);
					degraded.set(true);
					if (nodesInCluster == 2) {
						/*Redistribute signals*/
						List<OwnerBackupSignal> signalsOwnedList = new ArrayList<OwnerBackupSignal>();
						for (int i = 0; i < signalsToSendNumber; i++) {
							OwnerBackupSignal signalToSend = mySignalsList.get(i);
							/* Remove the signal from my signals and update the backup key*/
							List<OwnerBackupSignal> signalsListToModify = mySignals.get(address);
							signalsListToModify.remove(signalToSend);

							/* Update the owner of the signal to send */
							signalToSend.setOwner(destNode);

							/*Update the owner of the backuped signal that i have*/
							List<OwnerBackupSignal> myBackupsList = signalsIBackup.get(address);

							myBackupsList.remove(signalToSend);
							/* Change the key of the map */
							if(signalsIBackup.get(destNode) == null){
								List<OwnerBackupSignal> newList = new ArrayList<OwnerBackupSignal>();
								newList.add(signalToSend);
								signalsIBackup.put(destNode, newList);
							}
							else{
								List<OwnerBackupSignal> list = signalsIBackup.get(destNode);
								list.add(signalToSend);
							}

							signalsOwnedList.add(signalToSend);
						}
						/*Redistribute backups*/
						List<OwnerBackupSignal> signalsBackupedList = new ArrayList<OwnerBackupSignal>();
						for(int i=signalsToSendNumber; i<mySignalsList.size(); i++) {
							OwnerBackupSignal signalToSend = mySignalsList.get(i);
							/*Remove the signal from the backup*/
							List<OwnerBackupSignal> myBackupList = signalsIBackup.get(address);
							myBackupList.remove(signalToSend);

							/*Update the backup of the signal to send*/
							signalToSend.setBackup(destNode);

							/*Update the backup of the signal that I have*/
							List<OwnerBackupSignal> mySignalList = mySignals.get(address);
							mySignalList.remove(signalToSend);
							/* Change the key of the map */
							if(mySignals.get(destNode) == null){
								List<OwnerBackupSignal> newList = new ArrayList<OwnerBackupSignal>();
								newList.add(signalToSend);
								mySignals.put(destNode, newList);
							}
							else{
								List<OwnerBackupSignal> list = mySignals.get(destNode);
								list.add(signalToSend);
							}

							signalsBackupedList.add(signalToSend);

						}
						OwnerBackupSignalList obsl = new OwnerBackupSignalList();
						obsl.setOwnedSignals(signalsOwnedList);
						obsl.setBackupedSignals(signalsBackupedList);
						MultipleSignalsMessage message = new MultipleSignalsMessage(MsgOptions.ADD_MULTIPLE_SIGNALS, obsl);
						try {
							channel.send(destNode, message);
						} catch (Exception e) {
							e.printStackTrace();
						}
					}
					else {
						OwnerBackupSignalList obsl = new OwnerBackupSignalList();
						List<OwnerBackupSignal> signalsToSend = new ArrayList<OwnerBackupSignal>();
						for (int i = 0; i < signalsToSendNumber; i++) {
							OwnerBackupSignal signalToSend = mySignalsList.get(i);
							/* Remove the signal from my signals */
							System.out.println("My signal: " + signalToSend);

							List<OwnerBackupSignal> signalsListToModify = mySignals.get(signalToSend.getBackup());
							signalsListToModify.remove(signalToSend);

							/* Update the owner of the signal to send */
							signalToSend.setOwner(destNode);
							signalsToSend.add(signalToSend);

							/*Send the updated signal to the node who backups it*/
							OneSignalMessage updateOwner = new OneSignalMessage(MsgOptions.UPDATE_OWNER, signalToSend);
							try {
								channel.send(signalToSend.getBackup(), updateOwner);
							} catch (Exception e1) {
								e1.printStackTrace();
							}
						}

						obsl.setOwnedSignals(signalsToSend);
						MultipleSignalsMessage message = new MultipleSignalsMessage(MsgOptions.ADD_MULTIPLE_SIGNALS, obsl);
						try {
							channel.send(destNode, message);
						} catch (Exception e) {
							e.printStackTrace();
						}
					}

					try {
						degradedLatchForNewNode.await();
					} catch (InterruptedException e2) {
						e2.printStackTrace();
					}
					System.out.println("THE END OF DISTRIBUTION " + degraded.get());
					degraded.set(false);
				}
				System.out.println("Finish");
			}
		}.start();
	}

	
	/***********************************************************************************************/
	/*********************************************Messages******************************************/
	
	private class ReceiveProcess implements Runnable {
		@Override
		public void run() {
			while(true){
				Message msg = null;
				try {
					msg = receivedMessagesQueue.take();
				} catch (InterruptedException e) {
					return;
				}
				final MainMessage myMsg = (MainMessage) msg.getObject();
				MsgOptions option = myMsg.getOption();
				switch(option){
				case GIVE_ME_SIGNALS:
					sendSignalsToNewNode(msg.getSrc());
					break;
				case ADD_SIGNAL:
					addSignalProcess(msg);
					break;
				case ADD_SIGNAL_BACKUP:
					addSignalBackupProcess(msg);
					break;
				case ADD_MULTIPLE_SIGNALS:
					final Message myMessage = msg;
					new Thread(){
						public void run() {

							addMultipleSignalsProcess(myMessage);
							/*If a node is sending signals to the new node*/
							/*If I am the new node send the oks altogether*/
							if(imNew.get() == true){
								newNodeLatch.countDown();
								try {
									System.out.println("ADD_MULTIPLE received");
									newNodeLatch.await();
									System.out.println("LATCH FINISHED *");
								} catch (InterruptedException e) {
									e.printStackTrace();
								}
								imNew.set(false);
								degraded.set(false);
							}
							try {
								channel.send(myMessage.getSrc(), new OneSignalMessage(MsgOptions.ADD_MULTIPLE_SIGNALS_OK, null));
							} catch (Exception e) {
								e.printStackTrace();
							}
						};
					}.start();
					break;
				case ADD_MULTIPLE_SIGNALS_OK:
					degradedLatchForNewNode.countDown();
					distributeSignalsLatch.countDown();
					distributing.set(false);
					distributingOnNewNodeLatch.countDown();
					break;
				case UPDATE_OWNER:
					updateOwnerProcess(msg);
					break;
				case FIND_SIMILAR_SIGNAL:
					isInSearch.set(true);
					findSimilarSignalProcess(msg);
					break;
				case FIND_SIMILAR_SIGNAL_RESULT:
					findSimilarSignalResultProcess(msg);
					break;
				case DEGRADED:
					degraded.set(true);
					break;
				case NOT_DEGRADED:
					System.out.println("Sending NOT_DEGRADED");
					/*degradedLatch is useless in the case of a new node connection*/
					degradedLatch.countDown();
					break;
				case COORDINATOR_NOT_DEGRADED:
					System.out.println("COORDINATOR_NOT_DEGRADED");
					latchToWaitForDegradation.countDown();
					System.out.println("isInSearch " + isInSearch.get());
					System.out.println("searchInterrupted " + searchInterrupted.get());
					if(isInSearch.get() == true && searchInterrupted.get() == true){
						System.out.println("Deprecated latch countdown");
						deprecatedLatch.countDown();
					}
					break;
				}
			}
		}
	}
	
	@Override
	public void receive(Message msg) {
		try {
			receivedMessagesQueue.put(msg);
		} catch (InterruptedException e1) {
			e1.printStackTrace();
		}
	}
	

	private synchronized void addSignalProcess(Message originalMsg) {
		OneSignalMessage myMsg = (OneSignalMessage) originalMsg.getObject();
		
		List<OwnerBackupSignal> signals = mySignals.get(myMsg.getSignal().getBackup());
	
		/*If the backup node didn't backup any signals from me yet*/
		if(signals == null) {
			List<OwnerBackupSignal> newSignals = new ArrayList<OwnerBackupSignal>();
			newSignals.add(myMsg.getSignal());
			mySignals.put(myMsg.getSignal().getBackup(), newSignals);
		}
		else{
			signals.add(myMsg.getSignal());
		}
	}
	
	private synchronized void addSignalBackupProcess(Message originalMsg) {
		OneSignalMessage myMsg = (OneSignalMessage) originalMsg.getObject();
		List<OwnerBackupSignal> signals = signalsIBackup.get(myMsg.getSignal().getOwner());
	
		/*If I didn't backup any signals from the owner yet*/
		if(signals == null) {
			List<OwnerBackupSignal> newSignals = new ArrayList<OwnerBackupSignal>();
			newSignals.add(myMsg.getSignal());
			signalsIBackup.put(myMsg.getSignal().getOwner(), newSignals);
			
		}
		else{
			signals.add(myMsg.getSignal());
		}
	}
	
	private synchronized void updateOwnerProcess(Message originalMsg) {
		/*The owner of the signal is the one who sends the msg*/
		OneSignalMessage myMsg = (OneSignalMessage) originalMsg.getObject();
		List<OwnerBackupSignal> signalsIBackupList = signalsIBackup.get(originalMsg.getSrc());
		signalsIBackupList.remove(myMsg.getSignal());
		addSignalBackupProcess(originalMsg);
	}
	
	private void findSimilarSignalResultProcess(Message originalMsg){
		ResultMessage myMsg = (ResultMessage) originalMsg.getObject();
		try {
			resultList.put(myMsg.getResult());
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		latch.countDown();
	}
	
	
	private synchronized void addMultipleSignalsProcess(Message originalMsg) {
		MultipleSignalsMessage myMsg = (MultipleSignalsMessage) originalMsg.getObject();
		OwnerBackupSignalList obsl = myMsg.getSignalsToAdd();
		List<OwnerBackupSignal> obsOwn = obsl.getOwnedSignals();
		List<OwnerBackupSignal> obsBackup = obsl.getBackupedSignals();
		
		System.out.println("OWNS* " + obsOwn.size());
		System.out.println("BACKS* " + obsBackup.size());
		
		for(OwnerBackupSignal signal : obsOwn){
			originalMsg.setObject(new OneSignalMessage(null, signal));
			addSignalProcess(originalMsg);
		}
		
		for(OwnerBackupSignal signal : obsBackup){
			originalMsg.setObject(new OneSignalMessage(null, signal));
			addSignalBackupProcess(originalMsg);
		}
		
		System.out.println("FINISHED adding");
	}
	
	
	private synchronized void findSimilarSignalProcess(final Message originalMsg) {
		new Thread(){
			@Override
			public void run() {
				final OneSignalMessage myMsg = (OneSignalMessage) originalMsg.getObject();
				signalsToFind = new LinkedBlockingQueue<SignalsToFind>();
				whoStartedSearch = originalMsg.getSrc();
				signalsExecutor = Executors.newFixedThreadPool(threadsNumber);
				resultsQueue = new LinkedBlockingQueue<Result>();
				Result ans = new Result(myMsg.getSignal().getSignal());

				/*SignalsProcessor constructor*/
				//TODO use a static variable instead of the class SignalsToFind
				List<OwnerBackupSignal> signals = mapToList(mySignals);
				for(OwnerBackupSignal signal : signals){
					try {
						signalsToFind.put(new SignalsToFind(myMsg.getSignal().getSignal(), signal.getSignal()));
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
				}

				for(int i=0; i<threadsNumber; i++){
					signalsExecutor.execute(new SignalsProcessor());
				}

				try {
					signalsExecutor.shutdown();
					signalsExecutor.awaitTermination(signals.size() * 3 + 100, TimeUnit.SECONDS);
				} catch (InterruptedException e1) {
					e1.printStackTrace();
				}

				int j=0;
				for(Result result : resultsQueue){
					System.out.println("Inside search, results: " + resultsQueue.size());
					for(Item item : result.items()){
						j++;
						System.out.println("Inside search, items: " + j);
						ans = ans.include(item);
					}
				}		

				ResultMessage message = new ResultMessage(MsgOptions.FIND_SIMILAR_SIGNAL_RESULT, ans);
				try {
					channel.send(originalMsg.getSrc(), message);
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		}.start();
	}
	/***********************************************************************************************/
	/***********************************************************************************************/
	
	
	/***********************************************************************************************/
	/**************************************Auxiliar methods*****************************************/
	
	private synchronized List<OwnerBackupSignal> mapToList(ConcurrentHashMap<Address, List<OwnerBackupSignal>> map){
		List<OwnerBackupSignal> ans = new ArrayList<OwnerBackupSignal>();
		Iterator<Entry<Address, List<OwnerBackupSignal>>> iterator = map.entrySet().iterator();
		while(iterator.hasNext()){
			ans.addAll(iterator.next().getValue());
		}
		return ans;
	}
	
	public void findSimilarToThread(final Signal signal) throws InterruptedException {
		resultList = new LinkedBlockingQueue<Result>();
		System.out.println("Executing search");
		System.out.println("FIND, MEMBERS IN CLUSTER " + channel.getView().getMembers().size());
		latch = new CountDownLatch(channel.getView().getMembers().size());
		OneSignalMessage message = new OneSignalMessage(MsgOptions.FIND_SIMILAR_SIGNAL,
				new OwnerBackupSignal(signal, null, null));
		try {
			channel.send(null, message);
			/*latch that waits for all channels to be sent*/
		} catch (Exception e1) {
			e1.printStackTrace();
		}
		try {
			latch.await();
		} catch (InterruptedException e) {
			System.out.println("LATCH INTERRUPTED");
			e.printStackTrace();
			signalsExecutor.shutdownNow();
			throw new InterruptedException();
		}
	}
	
	private Result processAllResults(Signal signal) {
		System.out.println("Processing results...");
		Result ans = new Result(signal);
		int i = 0;
		for(Result result : resultList){
			System.out.println("Result list " + resultList.size());
			for(Item item : result.items()){
				ans = ans.include(item);
				i++;
				System.out.println(i);
			}
		}		
		return ans;
	}
	
	/***********************************************************************************************/
	/***********************************************************************************************/
	
	
	/***********************************************************************************************/
	/***********************************Interfaces methods to implement*****************************/

	@Override
	public void join(String clusterName) throws RemoteException {
		try {
			channel = new JChannel();
			channel.setReceiver(this);
			channel.connect(clusterName);
			address = channel.getAddress();
			System.out.println("ADDRESS IN JOIN " + channel.getAddress());
			signalsNodeAlone = new LinkedBlockingQueue<Signal>();
		} catch (Exception e) {
			System.out.println("Error creating cluster");
			e.printStackTrace();
		}
		viewAcceptedThread = new Thread(new ViewAcceptedProcess());
		dequeingSignalThread = new Thread(new DequeingSignalProcess());
		receiveMessageThread = new Thread(new ReceiveProcess());
		viewAcceptedThread.start();
		dequeingSignalThread.start();
		receiveMessageThread.start();
		System.out.println("Cluster created");
	}


	@Override
	public void exit() throws RemoteException {
		/*Empties everything to let the node run alone*/
		comparisonSignals.set(0);
		mySignals = null;
		signalsIBackup = null;
		signalsNodeAlone = new LinkedBlockingQueue<Signal>();
		if(channel != null){
			viewAcceptedThread.interrupt();
			dequeingSignalThread.interrupt();
			receiveMessageThread.interrupt();
			channel.close();
		}
		channel = null;
		System.out.println("Node disconnected");
	}


	@Override
	public NodeStats getStats() throws RemoteException {
		if(channel != null) {
			System.out.println("NEW? " + imNew);
			return new NodeStats(address.toString(), comparisonSignals.get(),
				mapToList(mySignals).size(), mapToList(signalsIBackup).size(), degraded.get());
		}
		else {			
			return new NodeStats(address.toString(), comparisonSignals.get(), signalsNodeAlone.size(), 0, degraded.get());
		}
	}

	
	@Override
	public void add(Signal signal) throws RemoteException {
		/*If the node is alone*/
		if(channel == null){
			signalsNodeAlone.add(signal);
			return;
		}
		else{
			arrivingSignalsQueue.add(signal);
		}
	}
	

	private Result findSimilarSignalProcessQueue(Signal signalToFind, BlockingQueue<Signal> signals) {
		signalsToFind = new LinkedBlockingQueue<SignalsToFind>();
		signalsExecutor = Executors.newFixedThreadPool(threadsNumber);
		resultsQueue = new LinkedBlockingQueue<Result>();
		Result ans = new Result(signalToFind);

		/*SignalsProcessor constructor*/
		for(Signal signal : signals){
			try {
				signalsToFind.put(new SignalsToFind(signalToFind, signal));
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}

		for(int i=0; i<threadsNumber; i++){
			signalsExecutor.execute(new SignalsProcessor());
		}

		try {
			signalsExecutor.shutdown();
			signalsExecutor.awaitTermination(signals.size() * 3 + 1000, TimeUnit.SECONDS);
		} catch (InterruptedException e1) {
			e1.printStackTrace();
		}

		int j=0;
		for(Result result : resultsQueue){
			System.out.println("Inside search, results: " + resultsQueue.size());
			for(Item item : result.items()){
				j++;
				System.out.println("Inside search, items: " + j);
				ans = ans.include(item);
			}
		}
		
		return ans; 
	}
	
	@Override
	public Result findSimilarTo(final Signal signal) throws RemoteException {
		System.out.println("Executing search...");
		/*Set me as the one who started the search*/
		whoStartedSearch = address;
		if(channel == null){
			comparisonSignals.getAndIncrement();
			Result result = findSimilarSignalProcessQueue(signal, signalsNodeAlone);
			return result;
		}
		
		if (signal == null) {
			throw new IllegalArgumentException("Signal cannot be null");
		}

		resultList = new LinkedBlockingQueue<Result>();
		
		comparisonSignals.getAndIncrement();

		completed.set(false);

		Result localQueueResult = null;
		if(!arrivingSignalsQueue.isEmpty()){
			System.out.println("arrivingSignalsSearch");
			localQueueResult = findSimilarSignalProcessQueue(signal, arrivingSignalsQueue);
			resultList.add(localQueueResult);
		}
		
		while(completed.get() == false) {
			System.out.println("AQUI!!!");
			searchThread = new Thread(){
				public void run() {
					try {
						findSimilarToThread(signal);
						completed.set(true);
					} catch (InterruptedException e1) {
						searchInterrupted.set(true);
						deprecatedLatch = new CountDownLatch(1);
						System.out.println("CLUSTER SIZE: " + channel.getView().getMembers().size());
						/*Wait for the cluster to be non deprecated*/
						try {
							deprecatedLatch.await();
							searchInterrupted.set(false);
						} catch (InterruptedException e) {
							e.printStackTrace();
						}
					}
				};
			};
			
			searchThread.start();
			
			/*Wait for the thread to end before processing results*/
			try {
				searchThread.join();
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
			
			
		}
		
		isInSearch.set(false);
		return processAllResults(signal);
	}


	/***********************************************************************************************/
	/***********************************************************************************************/

	
	/***********************************************************************************************/
	/*****************************************Runnables*********************************************/
	private class ViewAcceptedProcess implements Runnable {
		@Override
		public void run() {
			while(true)	{
				try {
					UpDownNode udn = null;
					if(imNew.get() == false)
						 udn = viewAcceptedQueue.take();
					if(imNew.get() == true){
						do{
							udn = viewAcceptedQueue.take();
						} while(udn.getMsg().equals(ViewAcceptedMsg.UP) && !udn.getNode().equals(address));
					}
					switch(udn.getMsg()){
						case UP:
							if(imNew.get() == true){
								/*Initialize redistribution*/
								degraded.set(true);
								if(channel.getView().getMembers().size() == 1){
									imNew.set(false);
									degraded.set(false);
								}
								for(Address node : channel.getView().getMembers()){
									if(!node.equals(address))
										System.out.println("ADDRESS " + address);
										try {
											distributingOnNewNodeLatch = new CountDownLatch(1);
											distributing.set(true);
											newNodeLatch = new CountDownLatch(channel.getView().getMembers().size()-1);
											channel.send(node, new OneSignalMessage(MsgOptions.GIVE_ME_SIGNALS, null));
										} catch (Exception e) {
											e.printStackTrace();
										}
								}
							}
							System.out.println("Sali");
						break;
						case DOWN:
							/*Stop search thread if a node falls*/
							if(address.equals(whoStartedSearch)){
								System.out.println("THREAD INTERRUPTED");
								searchThread.interrupt();
							}
							System.out.println("NODE DOWN");
							distributeSignals(udn.getNode());
						break;
					}
				} catch (InterruptedException e) {
				}
			}
		}
	}
	
	
	private class SignalsProcessor implements Runnable {

		@Override
		public void run() {
			System.out.println("SignalsToFind.peek() " + signalsToFind.size());
			SignalsToFind signalToFind = signalsToFind.peek();
			if(signalToFind != null){
				Result result = new Result(signalToFind.getSignalToFind());
				SignalsToFind signal;
				while((signal = signalsToFind.poll()) != null){
					Result.Item item = new Result.Item(signal.getSignalInQueue(), signal.getSignalToFind().findDeviation(signal.getSignalInQueue()));
					result = result.include(item);
				}
				System.out.println("RESULT SIZE " + resultsQueue.size());
				try {
					resultsQueue.put(result);
				} catch (InterruptedException e) {
					return;
				}
			}
			System.out.println("COMPLETED!");
		}
	}
	
	
	private class DequeingSignalProcess implements Runnable{
		@Override
		public void run() {
			while(true){
				
				Signal signal = null;
				try {
					signal = arrivingSignalsQueue.take();
				} catch (InterruptedException e) {
					return;
				}
				try { 		
					if(distributing.get() == true && channel.getView().getMembers().size() > 1) 		
						distributingOnNewNodeLatch.await(); 		
				} catch (InterruptedException e1) { 		
					// TODO Auto-generated catch block 		
					e1.printStackTrace(); 		
				}
				List<Address> nodesList = new ArrayList<Address>();
				nodesList.addAll(channel.getView().getMembers());
				/*Choose a node to be the owner*/
				int random = (int)(Math.random() * (nodesList.size()));		
				
				Address ownerNodeAddress = nodesList.get(random);
				
				nodesList = new ArrayList<Address>();
				nodesList.addAll(channel.getView().getMembers());
				/*Choose a node to be the backup different from the owner*/
				if(nodesList.size() > 1)
					nodesList.remove(ownerNodeAddress);
				random = (int)(Math.random() * (nodesList.size()));
				
				Address backupNodeAddress = nodesList.get(random);
				
				OwnerBackupSignal ownerBackupSignal = new OwnerBackupSignal(signal, ownerNodeAddress, backupNodeAddress);
				
				OneSignalMessage destinyMessage = new OneSignalMessage(MsgOptions.ADD_SIGNAL, ownerBackupSignal);
				OneSignalMessage backupMessage = new OneSignalMessage(MsgOptions.ADD_SIGNAL_BACKUP, ownerBackupSignal);
				try {
					channel.send(ownerNodeAddress, destinyMessage);
					channel.send(backupNodeAddress, backupMessage);
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		}
	}
	
	/***********************************************************************************************/
	/***********************************************************************************************/
}
